package com.lonni.rpc.core.register.zk;

import com.alibaba.fastjson2.JSON;
import com.lonni.rpc.core.model.RpcConfig;
import com.lonni.rpc.core.model.ZkConfig;
import com.lonni.rpc.core.register.model.Service;
import com.lonni.rpc.core.register.model.ServiceURL;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.ACLBackgroundPathAndBytesable;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.util.Assert;

import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * Curator(zookeeper client) utils
 *
 * @author shuang.kou
 * @createTime 2020年05月31日 11:38:00
 */
@Slf4j
public final class ZKUtils {

    /**
     * 连接失败等待重试时间
     */
    private static final int BASE_SLEEP_TIME = 1000;
    /**
     * 重试次数
     */
    private static final int MAX_RETRIES = 3;
    /**
     * 根路径
     */
    public static final String ZK_REGISTER_ROOT_PATH = "/lonni-rpc";
    /**
     * 提供者标识
     */
    public static final String PROVIDER = "/provider";


    private static CuratorFramework zkClient;
    private static final String DEFAULT_ZOOKEEPER_ADDRESS = "127.0.0.1:2181";

    private ZkConfig zkConfig;

    private ZKUtils() {
    }

    public static ZKUtils me = new ZKUtils();

    public CuratorFramework client() {
        return zkClient;
    }

    public void init(RpcConfig rpcConfig) {
        this.zkConfig = rpcConfig.getZookeeper();
        String ipaddress = StringUtils.isNotBlank(zkConfig.getConnectString()) ? zkConfig.getConnectString() : DEFAULT_ZOOKEEPER_ADDRESS;
        Integer retryCount = Objects.nonNull(zkConfig.getRetryCount()) ? zkConfig.getRetryCount() : MAX_RETRIES;
        Integer retryTimes = Objects.nonNull(zkConfig.getElapsedTimeMs()) ? zkConfig.getElapsedTimeMs() : BASE_SLEEP_TIME;

        zkClient = CuratorFrameworkFactory
                .newClient(ipaddress,
                        zkConfig.getSessionTimeoutMs(),
                        zkConfig.getConnectionTimeoutMs(),
                        new ExponentialBackoffRetry
                                (retryTimes, retryCount)
                );
        zkClient.start();
        //直接创建子节点
        try {
            this.createRootNode(ZK_REGISTER_ROOT_PATH);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }


    public boolean existNode(String path) {
        try {
            Stat stat = zkClient.checkExists().forPath(path);
            return stat != null;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 创建持久化酒店
     *
     * @param path
     * @throws Exception
     */
    public void createRootNode(String path) throws Exception {
        try {
            if (!this.existNode(path)) {
                zkClient.create()
                        .creatingParentContainersIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        .forPath(path);
            }

        } catch (Exception exception) {
            log.error("zk 注册节点异常", exception);
            throw exception;

        }
    }


    /**
     * 创建虚拟化子节点
     *
     * @param path
     * @param data json字符串
     * @throws Exception
     */
    public void createNode(String path, String data) throws Exception {
        if (this.existNode(path)) {
            this.deleteNode(path);
        }
        zkClient.create()
                .creatingParentContainersIfNeeded()
                .withMode(CreateMode.EPHEMERAL)
                .forPath(path, StringUtils.isNotBlank(data) ? data.getBytes(StandardCharsets.UTF_8) : null);

    }

    public String getData(String path) {
        byte[] bytes = null;
        String result = "";
        try {
            bytes = zkClient.getData().forPath(path);
            return new String(bytes, StandardCharsets.UTF_8);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }


    /**
     * 删除节点
     *
     * @param path
     * @return
     */
    public boolean deleteNode(String path) {
        try {
            zkClient.delete().forPath(path);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }


    public String getProviderPathForIpPort(ServiceURL url) {

        return ZKUtils.ZK_REGISTER_ROOT_PATH +
                ZKUtils.PROVIDER + "/"
                + url.getInterfaceName() + ":"
                + url.getGroup() + ":"
                + url.getVersion() + "/"
                + url.getIp() + ":"
                + url.getPort();
    }

    /**
     * 获取地址
     * 用于从节点获取数据使用
     *
     * @param service
     * @param ipport  已经拼接号的ip和端口
     * @return
     */
    public String getProviderPathForIpPort(Service service, String ipport) {

        return ZKUtils.ZK_REGISTER_ROOT_PATH +
                ZKUtils.PROVIDER + "/"
                + service.getInterfaceName() + ":"
                + service.getGroup() + ":"
                + service.getVersion() + "/"
                + ipport
                ;
    }

    /**
     * 获取提供者地址 只到版本节点
     *
     * @param url
     * @return
     */
    public String getProviderPath(ServiceURL url) {

        return ZKUtils.ZK_REGISTER_ROOT_PATH + ZKUtils.PROVIDER + "/"
                + url.getInterfaceName() + ":"
                + url.getGroup() + ":"
                + url.getVersion();
    }

    public String getProviderPath(String interfaceName, String group, String version) {

        return ZKUtils.ZK_REGISTER_ROOT_PATH + ZKUtils.PROVIDER + "/" + interfaceName + ":" + group + ":" + version;
    }


}
